package com.navinfo.platform.flink.cdc.streaming;


import com.navinfo.platform.flink.cdc.streaming.functions.JsonDebeziumDeserializationSchema;
import com.navinfo.platform.flink.cdc.streaming.functions.SynchronizeFunction;
import com.navinfo.platform.flink.cdc.util.ConfigUtils;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions;
import org.apache.flink.contrib.streaming.state.RocksDBOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zhaopeng
 * @date 2022/5/29
 * @desc
 */
public class MysqlCdcJob {

    private static final Logger logger = LoggerFactory.getLogger(MysqlCdcJob.class);


    public static void main(String[] args) throws Exception {
        logger.error(">>>>>>数据同步程序开始启动>>>>>>");
        //1.加载配置文件
        ParameterTool config = ConfigUtils.loadJarPathConfig(args, "huashen-cdc.properties");

        //2.创建环境,
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //3、设置检查点
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.setCheckpointStorage(config.get("flink.checkpoint.dir"));
        checkpointConfig.setCheckpointInterval(Time.minutes(config.getInt("flink.checkpoint.interval")).toMilliseconds());
        checkpointConfig.setCheckpointTimeout(Time.minutes(config.getInt("flink.checkpoint.timeout")).toMilliseconds());
        checkpointConfig.setMinPauseBetweenCheckpoints(Time.minutes(config.getInt("flink.checkpoint.between")).toMilliseconds());
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        //4、StateBackend配置，
        EmbeddedRocksDBStateBackend rocks = new EmbeddedRocksDBStateBackend(true);
        rocks.setNumberOfTransferThreads(RocksDBOptions.CHECKPOINT_TRANSFER_THREAD_NUM.defaultValue());
        rocks.setWriteBatchSize(RocksDBConfigurableOptions.WRITE_BATCH_SIZE.defaultValue().getBytes());
        env.setStateBackend(rocks);

        //5.创建cdc的source
        String[] tableList = config.get("source.jdbc.tableList").split(",");


//        JdbcIncrementalSource<String> mySqlSource = new MySqlSourceBuilder()
//                .hostname(config.get("source.jdbc.hostname"))
//                .port(config.getInt("source.jdbc.port"))
//                .serverTimeZone(config.get("source.jdbc.timezone"))
//                .databaseList(config.get("source.jdbc.databaseList"))
//                .username(config.get("source.jdbc.username"))
//                .password(config.get("source.jdbc.password"))
//                .tableList(tableList)
//                .deserializer(new JsonDebeziumDeserializationSchema())
//                .includeSchemaChanges(true) // output the schema changes as well
//                .build();

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(config.get("source.jdbc.hostname"))
                .port(config.getInt("source.jdbc.port"))
                .serverTimeZone(config.get("source.jdbc.timezone"))
                .databaseList(config.get("source.jdbc.databaseList"))
                .username(config.get("source.jdbc.username"))
                .password(config.get("source.jdbc.password"))
                .tableList(tableList)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .includeSchemaChanges(true) // output the schema changes as well
                .build();

        //6、数据同步和业务的Function
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // set 4 parallel source tasks
                .setParallelism(4)
                .process(new SynchronizeFunction(config))
                // use parallelism 1 for sink to keep message ordering
                .setParallelism(1);

        env.execute("huashen-cdc-job MySQL Snapshot + Binlog");
        logger.error(">>>>>>数据同步程序开始运行>>>>>>");
    }
}
